package com.happy.xiao.common.rabbitmq;

import com.fasterxml.jackson.core.type.TypeReference;
import com.happy.xiao.common.rabbitmq.bean.MessageSend;
import com.happy.xiao.common.rabbitmq.service.ICommonMessage;
import com.happy.xiao.common.rabbitmq.util.MessageUtil;
import com.happy.xiao.common.redis.IRedisService;
import com.happy.xiao.common.rabbitmq.db.service.IMessageFailLogService;
import com.happy.xiao.common.utils.StringUtil;
import com.happy.xiao.common.utils.Util;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.util.StringUtils;

import java.util.HashMap;
import java.util.Map;

/**
 * RabbitMq 公共配置
 */
@Configuration
public class RabbitmqConfig {

    private static final Logger log = LoggerFactory.getLogger(RabbitmqConfig.class);

    @Autowired
    private Environment env;

    @Autowired
    private CachingConnectionFactory connectionFactory;

    @Autowired
    IRedisService iRedisService;

    @Autowired
    ICommonMessage iCommonMessage;

    @Autowired
    IMessageFailLogService iMessageFailLogService;

    // 可靠性队列，最大过期时间，每次修改这个值，需要重新创建对应的队列，这里设置一个比较大的值，保证可靠性的消息，最大存活时间为24天
    // long类型的长度有限，超过24天之后。就溢出了。如果有超出24天的死信消息，有如下想法
    // 1. 先设置一个最大的时间的死信消息。。当消息过期之后。被listener监听了。这个时候，先判断
    public static final long max_delay_time_reliability_check_queue =  24 * 24 * 60 * 60 * 1000;

    // 消息发送失败时，保存到redis中消息失败次数的 redis前缀
    private static final String redis_prefix_message_send_fail = "rabbit-send-count";

    /**
     * 发送消息配置信息
     * @return
     */
    @Bean
    public RabbitTemplate rabbitTemplate() {
        connectionFactory.setPublisherConfirms(true);
        connectionFactory.setPublisherReturns(true);
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        rabbitTemplate.setMandatory(true);

        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean ack, String cause) {
                try {
                    if (ack) {
                        // 发送成功
                        log.info("消息发送成功" + StringUtil.time() + ":correlationData({}),ack({}),cause({})", correlationData, ack, cause);
                    } else {
                        // 发送失败
                        // 没有数据， 这个值，是在发送消息的时候，对message进行json转换 需要对Message中设置MessageId, 这样才能确保这个值是唯一的
                        if (correlationData == null) return;

                        String messageStr = correlationData.getId();
                        if (StringUtils.isEmpty(messageStr)) return;

                        MessageSend sendMessage = Util.JsonToObject(new TypeReference<MessageSend>() {}, messageStr);

                        if (sendMessage == null) return;

                        String messageId = sendMessage.getId();         // 这个是messageId
                        String exchange = sendMessage.getExchange();    // 消息交换器
                        String routingKey = sendMessage.getRoutingKey();// 路由key
                        String message = sendMessage.getMessage();      // 消息内容
                        int delayTime = sendMessage.getDelayTime();     // 延时时间

                        String expiration = null;                       // 消息过期时间
                        if (delayTime != 0 ) expiration = delayTime + "";

                        // 重发消息
                        resendMessage(rabbitTemplate, messageId, exchange, routingKey, message, expiration);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey, replyCode, replyText, message);
                try {
                    String messageId = message.getMessageProperties().getMessageId();
                    if (StringUtils.isEmpty(messageId)) return;

                    // 这里是设置的过期时间，有可能没有设置 null
                    String expiration = message.getMessageProperties().getExpiration();
                    // 这里是消息内容
                    String msgStr = MessageUtil.getMessageContent(message);

                    // 重发消息
                    resendMessage(rabbitTemplate, messageId, exchange, routingKey, msgStr, expiration);

                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });
        return rabbitTemplate;
    }


    // TODO 简单消息模型构建
    @Bean
    public TopicExchange simpleExchange() {
        return new TopicExchange(env.getProperty("simple.mq.exchange.name"), true, false);
    }

    @Bean(name = "simpleQueue")
    public Queue simpleQueue() {
        return new Queue(env.getProperty("simple.mq.queue.name"), true);
    }

    @Bean
    public Binding simpleBinding() {
        // 一个队列，对来自一个Exchange的消息感兴趣， 暗号是routingKey
        return BindingBuilder.bind(simpleQueue()).to(simpleExchange()).with(env.getProperty("simple.mq.routing.key.name"));
    }

    // TODO ------------------------------------------------------------------------------------------------------------
    // TODO 公共的死信队列
    @Bean(name = "deadQueue")
    public Queue deadQueue() {
        Map<String, Object> args = new HashMap<>();

        String temp = env.getProperty("common.dead.queue.name");
        args.put("x-dead-letter-exchange", env.getProperty("common.dead.exchange.name"));
        args.put("x-dead-letter-routing-key", env.getProperty("common.dead.routing.key.name"));
        args.put("x-message-ttl", 10000);

        return new Queue(temp, true, false, false, args);
    }

    //绑定死信队列-面向生产端
    @Bean
    public TopicExchange deadExchange() {
        return new TopicExchange(env.getProperty("common.produce.exchange.name"), true, false);
    }

    @Bean
    public Binding deadBinding() {
        return BindingBuilder
                .bind(deadQueue())
                .to(deadExchange())
                .with(env.getProperty("common.produce.routing.key.name"));
    }

    //创建并绑定实际监听消费队列
    @Bean(name = "deadRealQueue")
    public Queue deadRealQueue() {
        return new Queue(env.getProperty("common.produce.dead.real.queue.name"), true);
    }

    @Bean
    public TopicExchange deadRealExchange() {
        return new TopicExchange(env.getProperty("common.dead.exchange.name"), true, false);
    }

    @Bean
    public Binding deadRealBinding() {
        return BindingBuilder
                .bind(deadRealQueue())
                .to(deadRealExchange())
                .with(env.getProperty("common.dead.routing.key.name"));
    }

    // TODO ------------------------------------------------------------------------------------------------------------


    // TODO ------------------------------------------------------------------------------------------------------------
    /* TODO 用于保证消息可靠性的队列
     * 业务完整逻辑：
     *      在执行完业务操作之后，开始发送消息， 第一个消息是业务消息， 第二个消息是检查业务消息【保证业务消息正常消费，延时消息】
     *
     * 一个上游队列，  这个队列【发送方】，用于接受第二次消息的数据【延时消息，确认消息是否成功投递】
     * 一个下游队列，  这个队列【消费方】，用于接收第一次业务消息，消费后，重新再发一条确认消息
     *
     * 回调服务，监听这个两个队列
     *      1. 监听到消费服务重新发的消息之后。把消息保存在消息数据库，确定已经消费了
     *      2. 监听第二次检查的消息，如果检测到消息已经消费成功之后。则不做处理。没有消费成功。则进行补偿机制【通过RPC远程调用接口，重新发送消息】
     *
     */
    @Bean(name = "checkDeadQueue")
    public Queue checkDeadQueue() {
        Map<String, Object> args = new HashMap<>();
        String temp = env.getProperty("check.dead.queue.name");
        args.put("x-dead-letter-exchange", env.getProperty("check.dead.exchange.name"));
        args.put("x-dead-letter-routing-key", env.getProperty("check.dead.routing.key.name"));
        args.put("x-message-ttl", max_delay_time_reliability_check_queue);  // 最大存活时间，每次修改这个值，需要重新创建这个队列，不然会报错
        return new Queue(temp, true, false, false, args);
    }

    //绑定死信队列-面向生产端
    @Bean
    public TopicExchange checkDeadExchange() {
        return new TopicExchange(env.getProperty("check.produce.exchange.name"), true, false);
    }

    @Bean
    public Binding checkDeadBinding() {
        return BindingBuilder
                .bind(checkDeadQueue())
                .to(checkDeadExchange())
                .with(env.getProperty("check.produce.routing.key.name"));
    }

    //创建并绑定实际监听消费队列
    @Bean(name = "checkDeadRealQueue")
    public Queue checkDeadRealQueue() {
        return new Queue(env.getProperty("check.produce.dead.real.queue.name"), true);
    }

    @Bean
    public TopicExchange checkDeadRealExchange() {
        return new TopicExchange(env.getProperty("check.dead.exchange.name"), true, false);
    }

    @Bean
    public Binding checkDeadRealBinding() {
        return BindingBuilder
                .bind(checkDeadRealQueue())
                .to(checkDeadRealExchange())
                .with(env.getProperty("check.dead.routing.key.name"));
    }

    // TODO: 消费者确认消息 模型构建
    @Bean
    public DirectExchange checkConfirmExchange() {
        return new DirectExchange(env.getProperty("check.confirm.exchange.name"), true, false);
    }

    @Bean(name = "checkConfirmQueue")
    public Queue checkConfirmQueue() {
        return new Queue(env.getProperty("check.confirm.queue.name"), true);
    }

    @Bean
    public Binding checkConfirmBinding() {
        // 一个队列，对来自一个Exchange的消息感兴趣， 暗号是routingKey
        return BindingBuilder.bind(checkConfirmQueue()).to(checkConfirmExchange()).with(env.getProperty("check.confirm.routing.key.name"));
    }

    // TODO ------------------------------------------------------------------------------------------------------------


    /**
     * 重发消息
     *
     * @param rabbitTemplate
     * @param messageId
     * @param exchange
     * @param routingKey
     * @param message
     * @param expiration  过期时间
     */
    private void resendMessage(RabbitTemplate rabbitTemplate,
                               String messageId,
                               String exchange,
                               String routingKey,
                               String message,
                               String expiration) {

        try {
            if (StringUtils.isEmpty(messageId)
                    || StringUtils.isEmpty(exchange)
                    || StringUtils.isEmpty(routingKey)) return;

            int delayTime = 0;


            if (!StringUtils.isEmpty(expiration)) {
                // 这里是延时消息处理
                delayTime = Integer.parseInt(expiration);
            }

            // 1. 把这个消息保存在redis中【重复发送7次，第7次之后，就不再发送了，然后，保存到本地数据库】
            // 这是这个消息的 redis的失败id
            String redisKey = redis_prefix_message_send_fail + messageId;
            Object keyValue = iRedisService.getKeyValue(redisKey);

            int messageFailCount = 1;

            // 存在这个messageId的失败次数
            if (keyValue != null) messageFailCount = (int) keyValue;

            if (messageFailCount > 7) {
                // 重发次数大于7之后，不再发送，而是保存到数据库
                System.out.println("发送了7次啦  要保存数据库了 " + messageFailCount + "   redisKey= " + redisKey);

                // 保存异常发送数据到数据库
                iMessageFailLogService.createFailMessageInDB(messageId, exchange, routingKey, message, expiration, 0);
                iRedisService.delete(redisKey);

            } else {
                // 重发送的消息
                if (delayTime <= 0) {
                    // 一般消息处理
                    iCommonMessage.sendNormal(rabbitTemplate, exchange, routingKey, message, messageId);
                } else {
                    // 延时消息处理
                    iCommonMessage.sendDelay(rabbitTemplate, exchange, routingKey, message, messageId, delayTime);
                }
                // 这里发送失败次数自增
                iRedisService.increment(redisKey, 1, 7 * 86400);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}
